package com.springboot.kafka.kafka;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * 监听服务器上的kafka是否有相关的消息发过来
 */
@Component
public class ConsumerTopicA {

    private final Log log = LogFactory.getLog(this.getClass());

    /**
     * 定义此消费者接收topics = "topicA"的消息
     *
     * @param record 变量代表消息本身，可以通过ConsumerRecord<?,?>类型的record变量来打印接收的消息的各种信息
     */
    @KafkaListener(topics = {"topicA"})
    public void listen(ConsumerRecord<?, ?> record) {
        log.info(String.format("<ConsumerTopicA> topic is %s, offset is %d, key: %s, value is %s", record.topic(), record.offset(), record.key(), record.value()));
    }
}
